package com.huitone.smspfm.socket.core;

import java.io.InputStream;
import java.util.Iterator;
import java.util.concurrent.ArrayBlockingQueue;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import com.huitone.smspfm.socket.model.AccountInfo;
import com.huitone.smspfm.socket.protocol.sgip.constant.CommandID;
import com.huitone.smspfm.socket.protocol.sgip.constant.ErrorCode;
import com.huitone.smspfm.socket.protocol.sgip.message.Message;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.ActiveTest;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.ActiveTestResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Bind;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.BindResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Deliver;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.DeliverResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report2;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Report2Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.ReportResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit2;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.Submit2Resp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.SubmitResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.UnBind;
import com.huitone.smspfm.socket.protocol.sgip.message.msgbody.UnBindResp;
import com.huitone.smspfm.socket.protocol.sgip.message.msghead.Head;
import com.huitone.smspfm.socket.server.SocketHolderManager;

/**   
* @Title MessageDispacher.java 
* @Package com.huitone.smspfm.socket 
* @Description TODO(用一句话描述该文件做什么) 
* @author ybliang   
* @date 2017年3月25日 下午9:04:32 
* @version V1.0   
*/
public class MessageDispacher {
	
	private static final Logger logger = LogManager.getLogger(MessageDispacher.class);
	
	private boolean isMaxFlux = false;
	
	/**
	 * 实际处理流量
	 */
	private int curFlux = 0;
	
	private int lastFlux = 0;
	
	/**
	 * 上一次流量控制开始时间
	 */
	private long lastTime = 0;

	public void readMessage(SocketHolder socketHolder) throws Exception {
		
		long startTime = System.currentTimeMillis();
		
		InputStream inputStream = socketHolder.getAccept().getInputStream();
		
		Head head = null;
		byte[] body = null;
		
		ArrayBlockingQueue<Byte> queue = new ArrayBlockingQueue<Byte>(Head.HEAD_LENGTH);
		
		int index = 0;
		
		while(true) { // 读取消息头(程序不断循环直到读到正确的消息头)
			
			queue.put((byte) inputStream.read());
			index ++;
			
			if (index == Head.HEAD_LENGTH) {
				
				boolean disconnected = true;
				
				int n = 0;
				byte[] bytes = new byte[Head.HEAD_LENGTH];
				
				Iterator<Byte> iterator = queue.iterator();
				while(iterator.hasNext() && n < Head.HEAD_LENGTH) {
					byte bytz = (byte) iterator.next();
					bytes[n] = bytz;
					n ++;
					if (bytz != -1) {
						disconnected = false;
					}
				}
				
				if (disconnected) {
					socketHolder.close();
					return;
				}
				
				head = Head.build(bytes);
				if (!CommandID.hasCommandID(head.getCommandID())) {
					queue.take();
					index --;
				} else {
					break;
				}
			}
		}
		
		if (head != null) {
			
			int bodyLength = head.getMessageLength() - Head.HEAD_LENGTH;
			
			body = new byte[bodyLength];
			
			for (int i = 0 ; i < bodyLength ; i ++) {
				body[i] = (byte) inputStream.read();
			}
			
			dispach(head, body, socketHolder);
		} else {
			
			logger.error("[空消息头错误]");
		}
		
		long endTime = System.currentTimeMillis();
		
		long time = endTime - startTime;
		
		logger.debug("读取一条消息耗时:" + time + "ms");
	}
	
	public void dispach(Head msgHead, byte[] body, SocketHolder socketHolder) {
		
		IMessageHandler messageHandler = socketHolder.getMessageHandler();
		
		switch (CommandID.getCommandIDem(msgHead.getCommandID())) {
		
		case SGIP_BIND:
			
			IUserChecker userChecker = socketHolder.getUserChecker();
			
			logger.info("校验" + msgHead.toString());
			
			if (userChecker != null) {
				
				Bind bind = Bind.build(body);
				
				String srvNode = bind.getSrvNode();
				String username = bind.getLoginName();
				String password = bind.getLoginPassword();
				
				boolean bindSuccess = false;
				
				int result = ErrorCode.CODE0;
				
				AccountInfo accountInfo = userChecker.isUserLogin2(srvNode, username, password);
				
				if (accountInfo != null) {
					
					socketHolder.setAccountInfo(accountInfo);
					
					switch (SocketHolderManager.getInstance().bind(socketHolder)) {
					
						case SUCCESS:
							logger.info("业务端:" + srvNode + "请求绑定,提示:绑定成功");
							bindSuccess = true;
							result = ErrorCode.CODE0;
							break;
						case MXFAILED:
							logger.info("业务端:" + srvNode + "请求绑定,提示:绑定失败,连接的业务端数超过了最大值:" + SocketHolderManager.getInstance().getMaxActive());
							result = ErrorCode.CODE_12;
							break;
						case MXSRVNODE:
							logger.info("业务端:" + srvNode + "请求绑定,提示:绑定失败,相同业务端打开的连接数超过了最大值:" + SocketHolderManager.getInstance().getMaxSameSrvNodeCount());
							result = ErrorCode.CODE_13;
							break;
						case BDFAILED:
							logger.info("业务端:" + srvNode + "请求绑定,提示:绑定失败,已经存在,不能重复绑定");
							result = ErrorCode.CODE_14;
							break;
						case ADDFAILED:
							logger.info("业务端:" + srvNode + "请求绑定,提示:绑定失败,添加不成功");
							result = ErrorCode.CODE_15;
							break;
						case NOLOGIN:
							logger.info("业务端:" + srvNode + "请求绑定,提示:没有登录");
							result = ErrorCode.CODE_16;
							break;
						default:
							break;
					}
					
				} else {
					result = ErrorCode.CODE_17;
					logger.info("业务端:" + srvNode + "请求绑定,提示:用户名密码校验失败");
				}
				
				response(msgHead, BindResp.class, result, CommandID.SGIP_BIND_RESP, socketHolder);
				
				if (!bindSuccess) {
					socketHolder.close();
				}
			} else {
				messageHandler.handleBindMessage(msgHead, Bind.build(body), socketHolder);
			}
			
			break;
			
		case SGIP_BIND_RESP:
			
			messageHandler.handleBindRespMessage(msgHead, BindResp.build(body));
			
			break;
			
		case SGIP_SUBMIT:
			
			fluxControll(msgHead, SubmitResp.class, CommandID.SGIP_SUBMIT_RESP, socketHolder);
			if (!isMaxFlux) {
				messageHandler.handleSubmitMessage(msgHead, Submit.build(body), socketHolder);
			}
			
			break;
								
		case SGIP_SUBMIT_RESP:
			
			messageHandler.handleSubmitRespMessage(msgHead, SubmitResp.build(body));
			
			break;
			
		case SGIP_DELIVER:
			
			messageHandler.handleDeliverMessage(msgHead, Deliver.build(body), socketHolder);
			
			break;
			
		case SGIP_DELIVER_RESP:
			
			messageHandler.handleDeliverRespMessage(msgHead, DeliverResp.build(body));
			
			break;
			
		case SGIP_REPORT:
			
			messageHandler.handleReportMessage(msgHead, Report.build(body), socketHolder);
			
			break;
			
		case SGIP_REPORT_RESP:
			
			messageHandler.handleReportRespMessage(msgHead, ReportResp.build(body));
			
			break;
			
		case SGIP_UNBIND:
			
			messageHandler.handleUnBindMessage(msgHead, UnBind.build(body), socketHolder);
			
			break;
			
		case SGIP_UNBIND_RESP:
			
			messageHandler.handleUnBindRespMessage(msgHead, UnBindResp.build(body));
			
			break;
			
		case SGIP_ACTIVE_TEST:
			
			messageHandler.handleActiveTestMessage(msgHead, ActiveTest.build(body), socketHolder);
			
			break;
			
		case SGIP_ACTIVE_TEST_RESP:
	
			messageHandler.handleActiveTestRespMessage(msgHead, ActiveTestResp.build(body));
	
			break;
		case SGIP_SUBMIT2:
			
			fluxControll(msgHead, Submit2Resp.class, CommandID.SGIP_SUBMIT2_RESP, socketHolder);
			if (!isMaxFlux) {
				messageHandler.handleSubmit2Message(msgHead, Submit2.build(body), socketHolder);
			}
			
			break;
		case SGIP_SUBMIT2_RESP:
			
			messageHandler.handleSubmit2RespMessage(msgHead, Submit2Resp.build(body));
	
			break;
			
		case SGIP_REPORT2:
			
			messageHandler.handleReport2Message(msgHead, Report2.build(body), socketHolder);
			
			break;
		case SGIP_REPORT2_RESP:
			
			messageHandler.handleReport2RespMessage(msgHead, Report2Resp.build(body));
			
			break;
	
		default:
			
			logger.debug("没有匹配到信息处理方法");
			
			break;
		}
	}
	
	/**
	 * @param msgHead 接收的消息头
	 * @param clazz 响应信息的body类
	 * @param commandID 命令ID
	 * @param socketHolder 通道连接
	 * @return true 通道达到了最大流量限制, false 通道流量还没有达到最大流量限制
	 */
	public synchronized void fluxControll(Head msgHead, Class<? extends Resp> clazz, long commandID, SocketHolder socketHolder) {
		
		if (socketHolder.getAccountInfo().isControlFlux() && socketHolder.getAccountInfo().isLogin()) {  // 流量控制
			
			if (lastTime == 0) {
				lastTime = socketHolder.getLastReadTimeMillis();
			}
			
			if (!isMaxFlux) {
				curFlux ++;
			}
			
			long interval = socketHolder.getLastReadTimeMillis() - lastTime;
			
			if (interval < 1000) {
				if (curFlux >= socketHolder.getAccountInfo().getMaxFlux()) {
					logger.debug("业务端" + socketHolder.getAccountInfo().getSrvNode() + "当前流量:" + curFlux + "(条/秒), 最大流量:" + socketHolder.getAccountInfo().getMaxFlux() + "(条/秒)");
					lastFlux = curFlux;
					isMaxFlux = true;
					response(msgHead, clazz, ErrorCode.CODE7, commandID, socketHolder);
				}
			} else {
				logger.debug("业务端" + socketHolder.getAccountInfo().getSrvNode() + "当前流量:" + curFlux + "(条/秒), 最大流量:" + socketHolder.getAccountInfo().getMaxFlux() + "(条/秒)");
				lastFlux = curFlux;
				curFlux = 0;
				lastTime = 0;
				isMaxFlux = false;
			}
		}
	}
	
	/**
	 * @param head 接收的消息头
	 * @param clazz 响应信息的body类
	 * @param result 接收消息的处理结果
	 * @param commandID 命令ID
	 * @param outputStream 客户端输出流
	 * @return 发送响应是否成功了
	 */
	public boolean response(Head head, Class<? extends Resp> clazz, int result, long commandID, SocketHolder socketHolder) {
		
		boolean respResult = true;
		
		try {
			
			Resp resp = clazz.newInstance();
			resp.setResult(result);
			
			Head respHead = Head.build(resp.length(), commandID, head.getSerialNoPart1(), head.getSerialNoPart2(), head.getSerialNoPart3());
			
			socketHolder.write(Message.build(respHead, resp));
			
		} catch (Exception e) {
			respResult = false;
			logger.error("登录响应信息报错.", e);
		}
		
		return respResult;
	}

	public int getCurFlux() {
		return curFlux;
	}

	public void setCurFlux(int curFlux) {
		this.curFlux = curFlux;
	}

	public int getLastFlux() {
		return lastFlux;
	}

	public void setLastFlux(int lastFlux) {
		this.lastFlux = lastFlux;
	}
}
